package io.mqttpush.getway.http;

import io.mqttpush.getway.GetWayConstantBean;
import io.mqttpush.getway.common.CommonResponse;
import io.mqttpush.getway.common.Statistics;
import io.mqttpush.getway.http.controller.ControllBeans;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 处理BC后端发过来的报文 接到报文并且丢给ab 端
 * 
 * @author tianzhenjiu
 *
 */
public class BcMqttHandle extends ChannelInboundHandlerAdapter {

	Logger logger = LoggerFactory.getLogger(getClass());

	/**
	 * 标志是否初始化了channnel
	 */
	static AtomicBoolean isInit = new AtomicBoolean(false);

	static final  Pattern pattern = Pattern.compile("http://(\\w+((\\.\\w+)*))(:(\\d+))?(/.*)?");

	/**
	 *
	 * 匹配host/端口/uri的组别
	 */
	static final  int patternGroups[]={1,5,6};


	final GetWayConstantBean constantBean = GetWayConstantBean.instance();

	public BcMqttHandle() {

		if (!isInit.get()) {
			if (isInit.compareAndSet(false, true)) {
				initAbChannel();
			}

		}
	}
	
	

	/**
	 * 初始化连接的参数
	 */
	public void initAbChannel() {

		Bootstrap bootstrap = constantBean.httpCallbackStap;
		bootstrap.group(constantBean.httpCallbackgroup).channel(NioSocketChannel.class)
				.handler(new ChannelInitializer<SocketChannel>() {
					@Override
					protected void initChannel(SocketChannel ch)  {

						ChannelPipeline pipeline=ch.pipeline();


						/**
						 * 超过五秒就关闭
						 */
						pipeline.addLast(new ReadTimeoutHandler(5, TimeUnit.SECONDS));

						pipeline.addLast(new HttpClientCodec());

						pipeline.addLast("chunkedWriteHandler", new ChunkedWriteHandler());
						/**
						 * 作用是将一个Http的消息组装成一个完成的HttpRequest或者HttpResponse，那么具体的是什么
						 * 取决于是请求还是响应, 该Handler必须放在HttpServerCodec后的后面
						 */
						pipeline.addLast("httpObjectAggregator", new HttpObjectAggregator(8192));

						pipeline.addLast(new HttpCallbackHandle());
					}

				});
	}

	/**
	 * 只处理两种消息，即登录成功和收到发布消息
	 */
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

		if (msg instanceof MqttConnAckMessage) {

			MqttConnAckMessage ackMessage = (MqttConnAckMessage) msg;

			switch (ackMessage.variableHeader().connectReturnCode()) {
			case CONNECTION_ACCEPTED:
				processLoginSuccess(ctx.channel());
				break;
			default:
				break;
			}
		} else if (msg instanceof MqttPublishMessage) {
			
			MqttPublishMessage publishMessage = (MqttPublishMessage) msg;
			replay(publishMessage,ctx);

			callbackMessage(ctx.channel(), publishMessage);

		} else {
			ctx.fireChannelRead(msg);
		}

	}

	
	/**
	 * 回复
	 * @param publishMessage
	 * @param ctx
	 */
	void replay(MqttPublishMessage publishMessage,ChannelHandlerContext ctx) {
		
	
		
		try {
			MqttPubAckMessage replMessage=null;
			MqttQoS mqttQoS=publishMessage.fixedHeader().qosLevel();
			
			if(mqttQoS.equals(MqttQoS.EXACTLY_ONCE)) {
				
				replMessage= new MqttPubAckMessage(
						new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_LEAST_ONCE, false, 0),
						MqttMessageIdVariableHeader.from(publishMessage.variableHeader().packetId()));
			
				
			}else if(mqttQoS.equals(MqttQoS.AT_LEAST_ONCE)) {
				
				replMessage= new MqttPubAckMessage(
						new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 0),
						MqttMessageIdVariableHeader.from(publishMessage.variableHeader().packetId()));
			}
			
			
			if(replMessage!=null) {
				ctx.writeAndFlush(replMessage);
			}
		} catch (Exception e) {
				logger.warn("回复异常",e);
		}
	}
	
	/**
	 * 回调http 接口消息
	 * 
	 * @param bcChannel
	 * @param publishMessage
	 */
	public void callbackMessage(Channel bcChannel, MqttPublishMessage publishMessage) {

		String callback = bcChannel.attr(constantBean.bcHttpCallBackAttr).get();

		if (callback != null) {

			/**
			 * 匹配出host 和端口
			 */


			Matcher matcher = pattern.matcher(callback);
			String host = null;
			String uri=null;
			int port = 80;
			if (matcher.find()) {

				String gt = matcher.group(patternGroups[0]);
				if (gt != null) {
					host = gt;
				}

				if (host != null && (gt = matcher.group(patternGroups[1])) != null) {
					port = Integer.parseInt(gt);
				}

				if(host!=null&&(gt=matcher.group(patternGroups[2]))!=null){
					uri=gt;
				}



				if(uri==null||uri.isEmpty()){
					uri="/";
				}

			}

			if (host != null) {

				/**
				 * 连接回调服务器 并在连接成功之后 发消息
				 */

				String hhost=host.concat(":").concat(String.valueOf(port));
				String huri=uri;


				ChannelFuture channelFuture = constantBean.httpCallbackStap.connect(host, port);
				final ByteBuf httpContent = publishMessage.content();
				channelFuture.addListener((ChannelFuture future) -> {
					if (future.isSuccess()) {
						
				
						
						httpContent.retain();
						
						DefaultFullHttpRequest remoteCallRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,
								HttpMethod.POST, huri, httpContent);

						remoteCallRequest.headers().set(CommonResponse.CONTENT_TYPE, CommonResponse.jsonType);
						remoteCallRequest.headers().set(CommonResponse.Host, hhost);
						remoteCallRequest.headers().set(CommonResponse.CONTENT_LENGTH, httpContent.readableBytes());
						remoteCallRequest.headers().set(CommonResponse.Accept, CommonResponse.AllAccept);

						/**
						 *
						 * 不要关闭，等5秒自动关闭或者收到response的时候关闭
						 */
						future.channel().writeAndFlush(remoteCallRequest);
						/**
						 * http回调数+1
						 */
						Statistics.httpResCount.incrementAndGet();
						
						if(logger.isDebugEnabled()) {
							
							byte[] bs=new byte[httpContent.readableBytes()];
							httpContent.readBytes(bs);
							logger.debug("调用callback{}->uri{}",callback,huri);
							logger.debug("content:{}",new String(bs));
						}


						if(httpContent.refCnt()>0){
							httpContent.release();
						}



					}else {
						logger.warn("连接回调地址失败{}",callback);
					}
				});
			}else{
				logger.warn("回调需要找到http主机，现在没有找到host{}",callback);
			}
		}else{

			logger.warn("为什么call地址为空?");
		}

	}

	/**
	 * 处理登录成功，发送缓存的消息
	 * 
	 * @param channel
	 */
	public void processLoginSuccess(Channel channel) {

		channel.attr(ControllBeans.loginKey).set(true);

		MqttPublishMessage mqttPublishMessages = null;

		while ((mqttPublishMessages = ControllBeans.mqttPublishMessages.poll()) != null) {
			channel.write(mqttPublishMessages);
		}
		
		if(logger.isDebugEnabled()) {
			logger.debug("登录成功,http写入MQTT");
		}

		channel.flush();
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		super.channelActive(ctx);
		/**
		 * http后端连接数+1
		 */
		Statistics.httpBcCount.incrementAndGet();
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		super.channelInactive(ctx);

		/**
		 * http后端连接数-1
		 */
		Statistics.httpBcCount.decrementAndGet();
	}
}
